package org.project.appmark

import org.FlinkStreamApp
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.project.bean.SimulatedEventSource

/**
 * description ：APP 不分渠道数据统计
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/27 11:39
 * modified By ：
 * version:    : 1.0
 */
object AppMarketingStatistics extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 获取源数据
    val source = environment.addSource(new SimulatedEventSource)

    //  APP 不分渠道数据统计
    source
      .assignAscendingTimestamps(_.ts * 1000)
      .map(ss => (ss.behavior, 1L))
      .keyBy(_._1)
      .timeWindow(Time.seconds(60), Time.seconds(5))
      .process(new MarketingCountTotal)
      .print()
  }

  class MarketingCountTotal extends ProcessWindowFunction[(String, Long), (String, Long), String, TimeWindow] {
    override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[(String, Long)]) = {
      out.collect(key, elements.size)
    }
  }

}
